package com.itzx.loginfail_detect


import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

// 输入的登录事件样例类
case class LoginEvent(userId: Long, ip: String, eventType: String, eventTime: Long)
// 输出的异常报警信息样例类
case class Warning(userId: Long, firstFailTime: Long, lastFailTime: Long, warningMsg: String)

/**
 *
 *
 * author: yyeleven
 * create: 2020/3/24 21:50
 */
object LoginFail {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    val resource = getClass.getResource("/LoginLog.csv")
    val loginEventStream = env.readTextFile(resource.getPath)
      .map(data => {
        val dataArray = data.split(",")
        LoginEvent(dataArray(0).trim.toLong, dataArray(1).trim, dataArray(2).trim, dataArray(3).trim.toLong)
      }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(5)) {
      override def extractTimestamp(element: LoginEvent): Long = element.eventTime * 1000L
    })

    val warningStream = loginEventStream
      .keyBy(_.userId)  // 以用户id做分组
      .process(new LoginWarning(2))

    warningStream.print()
    env.execute("login fail detect job")
  }
}


class LoginWarning(maxFailTimes: Int) extends KeyedProcessFunction[Long, LoginEvent, Warning] {

  // 定义状态, 保存2秒内的所有登录失败事件
  lazy val loginFailState: ListState[LoginEvent] = getRuntimeContext.getListState(new ListStateDescriptor[LoginEvent]("login-fail", classOf[LoginEvent]))


  override def processElement(value: LoginEvent, ctx: KeyedProcessFunction[Long, LoginEvent, Warning]#Context, out: Collector[Warning]): Unit = {
/*    val loginFailList = loginFailState.get()

    // 判断类型是否是fail, 只添加fail的事件到状态
    if (value.eventType == "fail") {
      if (!loginFailList.iterator().hasNext) {
        ctx.timerService().registerEventTimeTimer(value.eventTime * 1000L + 2000L)
      }
      loginFailState.add(value)
    } else {
      // 如果是成功,直接清空状态
      loginFailState.clear()
    }*/

    if (value.eventType == "fail") {
      // 如果是失败, 判断之前是否有登录失败事件
      val iter = loginFailState.get().iterator()
      if (iter.hasNext) {
        // 如果已经有登录失败事件, 就比较事件时间
        val firstFail = iter.next()
        if (value.eventTime < firstFail.eventTime + 2000L) {
          // 如果两次间隔小于2秒, 输出报警
          out.collect(Warning(value.userId, firstFail.eventTime, value.eventTime, "login fail in 2 seconds for " + 2 + " times"))
        }
        // 更新最近一次的登录失败事件,保存在状态里
        loginFailState.clear()
        loginFailState.add(value)
      } else {
        // 如果是第一次登录失败, 直接添加到状态
        loginFailState.add(value)
      }
    } else {
      // 如果是成功, 清空状态
      loginFailState.clear()
    }
  }

/*  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, LoginEvent, Warning]#OnTimerContext, out: Collector[Warning]): Unit = {
    // 触发定时器的时候,根据状态里的失败个数决定是否输出报警
    // 拿出状态中的所有数据
    val allLoginFails: ListBuffer[LoginEvent] = new ListBuffer[LoginEvent]()
    val iter = loginFailState.get().iterator()
    while (iter.hasNext) {
      allLoginFails += iter.next()
    }

    // 判断个数
    if (allLoginFails.length >= maxFailTimes) {
      out.collect(Warning(allLoginFails.head.userId, allLoginFails.head.eventTime, allLoginFails.last.eventTime, "login fail in 2 seconds for " + allLoginFails.length + " times"))
    }
    // 清空状态
    loginFailState.clear()
  }*/
}
